//
// Created by Suzhen on 2017/2/27.
//

#include "SocketIO.h"

USING_ORION_V2

SocketIO *SocketIO::CreateSocket(FdType fd,
                                 const IOBehavior &behavior,
                                 uint sockIdleTime,
                                 SSL *ssl) {
    return new SocketIO(fd, behavior, sockIdleTime, ssl);
}

SocketIO::SocketIO(FdType fd,
                   const IOBehavior &behavior,
                   uint sockIdleTime,
                   SSL *ssl)
        : FileStream(fd, behavior, sockIdleTime),
          sslCtx_(ssl) {
    //设置 SSL 安全套接字
    if (sslCtx_) {
        SSL_set_fd(sslCtx_, coreAsyncPollData.fd);
        SSL_set_accept_state(sslCtx_);
        SSL_set_mode(sslCtx_, SSL_MODE_RELEASE_BUFFERS);
    }
}

SocketIO::~SocketIO() {
    if (sslCtx_) {
        SSL_free(sslCtx_);
    }
#ifdef ENABLE_LOG
    LOG_I << "SocketIO dealloc";
#endif
}

void SocketIO::onStreamIO(int status, EventFlag ioe) {
    if (sslCtx_ != NULL) {
        TlsStreamIoCallBack(status, ioe);
    } else {
        NormalStreamIoCallBack(status, ioe);
    }
}

void SocketIO::NormalStreamIoCallBack(int status, EventFlag events) {
    SocketIO *s = this;
    AsyncEventGenerator *generator = s->eventGenerator_;
    AsyncPollData &pollData = s->coreAsyncPollData;
    if (status < 0) {
#ifdef ENABLE_LOG
        LOG_I << "clientIoCb status != 0 , client disconnected";
#endif
        s->closeAsync();//此时会立即关闭会话
        return;
    }
    if (events & IOEvent::WRITABLE) {
        if (s->dataLock_.try_lock()) {
            do {
                ssize_t sndLen = 0;
                std::list<IOBuffer> &dataWillSend = s->dataWillSend_;
                if (dataWillSend.empty()) {
                    //change events to READ
                    s->activate(IOEvent::READABLE);
                    //unlock send queue buf
                    s->dataLock_.unlock();
                    break;
                }
                IOBuffer &itm = dataWillSend.front();
                const RawDataType *bufWillSnd = itm.buf + itm.index;
                size_t bufWntSnd = itm.len - itm.index;
                sndLen = ::send(pollData.fd, bufWillSnd, bufWntSnd, OPT_NOT_SIG_PIPE);
                if (sndLen >= 0 && (size_t) sndLen == bufWntSnd) {//compare signed and unsigned
                    itm.index += sndLen;//数据已经发送完
                    //send success
                    itm.completePromise_->setValue(YES);
                    dataWillSend.pop_front();
#ifdef ENABLE_LOG
                    LOG_I << "send data len = " << sndLen;
#endif
                } else if (sndLen == -1 && errno != EAGAIN) { // -1 means occur some error ,check error no
                    //如果不是 EAGAIN , 则表示sock出现问题, 需要断开连接
#ifdef ENABLE_LOG
                    LOG_I << "sock error  , errCode = " << sndLen
                          << " errno = " << errno;
#endif
                    s->dataLock_.unlock();
                    s->closeAsync();//此命令不会触发async执行,会直接产生调用
                    break;
                } else {
                    itm.index += sndLen;//这里数据还没有发送完
                    //todo 这里可以实现"发送了多少数据"的回调
#ifdef ENABLE_LOG
                    LOG_I << "EAGAIN, some data not send , send data len = " << sndLen
                          << " bufWntSnd = " << bufWntSnd
                          << " remain data len = " << (bufWntSnd - sndLen);
#endif
                    s->dataLock_.unlock();
                    break;
                }
            } while (1);
        }
    }
    //
    if (events & IOEvent::READABLE) {
        RawDataType *buf = generator->getReadBuffer();
        size_t bufSz = generator->getReadBufferSize();
        ssize_t recvLen = ::recv(pollData.fd, buf, bufSz, 0);
        if (recvLen > 0) {
#ifdef ENABLE_LOG
            LOG_I << "data read , recvLen = " << recvLen;
#endif
            s->OnDataRead(buf, (size_t) recvLen);
        } else if (recvLen == 0) {
#ifdef ENABLE_LOG
            LOG_I << "client offline , got FIN";
#endif
            s->closeAsync();//此时会立即关闭会话
            return;
        } else if (recvLen == -1 && errno != EAGAIN) {
            //sock error
#ifdef ENABLE_LOG
            LOG_I << "client offline , sock error , recvLen = " << recvLen << " errno " << errno;
#endif
            s->closeAsync();//此时会立即关闭会话
            return;
        } /* else { got_eagain();}*/
    }
}

void SocketIO::TlsStreamIoCallBack(int status, EventFlag events) {
    SocketIO *s = this;
    AsyncEventGenerator *generator = s->eventGenerator_;
    AsyncPollData &pollData = s->coreAsyncPollData;
    SSL *ssl = s->sslCtx_;
    if (status < 0) {
#ifdef ENABLE_LOG
        LOG_I << "clientIoCb status != 0 , client disconnected";
#endif
        s->closeAsync();
        return;
    }
    if (!s->dataWillSend_.empty()
        && ((events & IOEvent::WRITABLE) || SSL_want(ssl) == SSL_READING)) {
        if (s->dataLock_.try_lock()) {
            do {
                ssize_t sndLen = 0;
                std::list<IOBuffer> &dataWillSend = s->dataWillSend_;
                if (dataWillSend.empty()) {
                    //change events READ
                    s->activate(IOEvent::READABLE);
                    s->dataLock_.unlock();
                    break;
                }
                IOBuffer &itm = dataWillSend.front();
                const RawDataType *bufWillSnd = itm.buf + itm.index;
                size_t bufWntSnd = itm.len - itm.index;
                sndLen = SSL_write(ssl, bufWillSnd, (int) bufWntSnd);
                if (sndLen >= 0 && (size_t) sndLen == bufWntSnd) { //compare signed and unsigned
                    itm.index += sndLen;//数据已经发送完
                    //send success
                    itm.completePromise_->setValue(YES);
                    dataWillSend.pop_front();
#ifdef ENABLE_LOG
                    LOG_I << "SSL send data len = " << sndLen;
#endif
                    if ((pollData.events & IOEvent::WRITABLE) && SSL_want(ssl) != SSL_WRITING) {
                        s->activate(IOEvent::READABLE);
                    }
                } else if (sndLen <= 0) {
                    s->dataLock_.unlock();
                    switch (SSL_get_error(ssl, (int) sndLen)) {
                        case SSL_ERROR_WANT_READ:
                            break;
                        case SSL_ERROR_WANT_WRITE:
                            if ((pollData.events & UV_WRITABLE) == 0) {
                                s->activate(pollData.events | IOEvent::WRITABLE);
                            }
                            break;
                        default:
                            s->closeAsync();
                            goto emptyCase;
                    }
                    break;
                }
            } while (1);
        }
    }
    //
    if (events & IOEvent::READABLE) {
        do {
            RawDataType *buf = generator->getReadBuffer();
            size_t bufSz = generator->getReadBufferSize();
            int length = SSL_read(ssl, buf, (int) bufSz);
            if (length <= 0) {
                int sslErrNo = SSL_get_error(ssl, length);
                switch (sslErrNo) {
                    case SSL_ERROR_WANT_READ:
                        break;
                    case SSL_ERROR_WANT_WRITE:
                        if ((pollData.events & UV_WRITABLE) == 0) {
                            s->activate(pollData.events | IOEvent::WRITABLE);
                        }
                        break;
                    default: {
#ifdef ENABLE_LOG
                        LOG_I << "SSL ERROR , sslErrNo = " << sslErrNo << " errno = " << errno;
#endif
                        s->closeAsync();
                        goto emptyCase;
                    }
                }
                break;
            } else {
                s->OnDataRead(buf, (size_t) length);
                if (s->isClosing_) {
                    goto emptyCase;
                }
            }
        } while (SSL_pending(ssl));
    }
    emptyCase:
    {};
}

void SocketIO::setTcpCork(int enable) {
    SYS_FAULT(isAttached()) << "尚未附加到server上,setTcpCork失败";
#if defined(TCP_CORK)
    // Linux & SmartOS have proper TCP_CORK
    setsockopt(coreAsyncPollData.fd, IPPROTO_TCP, TCP_CORK, &enable, sizeof(int));
#elif defined(TCP_NOPUSH)
    // Mac OS X & FreeBSD have TCP_NOPUSH
    setsockopt(coreAsyncPollData.fd, IPPROTO_TCP, TCP_NOPUSH, &enable, sizeof(int));
    if (!enable) {
        // Tested on OS X, FreeBSD situation is unclear
        ::send(coreAsyncPollData.fd, "", 0, OPT_NOT_SIG_PIPE);
    }
#endif
}

void SocketIO::setTcpNoDelay(int enable) {
    SYS_FAULT(isAttached()) << "尚未附加到EventGenerator中,此时不应该setTcpNoDelay";
    setsockopt(coreAsyncPollData.fd, IPPROTO_TCP, TCP_NODELAY, &enable, sizeof(int));
}
